/**
 * Copyright © 2011 Creativor Studio.
 * About license information, please see LICENSE.txt.
 */
package org.creativor.rayson.transport.client;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.creativor.rayson.transport.api.TimeLimitConnection;
import org.creativor.rayson.util.Log;

/**
 * This listener will listen all asynchronous read and write events happens in
 * all the RPC connection in client side.
 * 
 * @author Nick Zhang
 */
class Listener extends Thread
{
	private static Logger LOGGER = Log.getLogger();
	private ConnectionManager connectionManager;
	
	private Lock lock;
	
	private Condition nonePendingTasks;
	private long pendingTaskCount;
	
	private boolean running = true;
	
	private Selector selector;
	
	/**
	 * Instantiates a new listener.
	 * 
	 * @param connectionManager
	 *            the connection manager
	 */
	Listener(ConnectionManager connectionManager)
	{
		setName("Transport client listener");
		this.connectionManager = connectionManager;
		this.lock = new ReentrantLock();
		this.nonePendingTasks = lock.newCondition();
		this.pendingTaskCount = 0;
	}
	
	/**
	 * Accept a new connection.
	 * 
	 * @param socketChannel
	 *            Underling socket channel.
	 * @param clientConnection
	 *            The client connection to be accept.
	 * @return selection key register into the selector of this listener.
	 * @throws ClosedChannelException
	 *             If the underling socket channel is closed before the result
	 *             is return .
	 */
	SelectionKey accept(SocketChannel socketChannel, TimeLimitConnection clientConnection)
			throws ClosedChannelException
	{
		SelectionKey key;
		selector.wakeup();
		lock.lock();
		try
		{
			pendingTaskCount++;
			try
			{
				key = socketChannel.register(selector, SelectionKey.OP_READ);
				key.attach(clientConnection);
			} finally
			{
				pendingTaskCount--;
			}
			if (pendingTaskCount == 0) nonePendingTasks.signalAll();
		} finally
		{
			lock.unlock();
		}
		return key;
	}
	
	/**
	 * Handling read event of this listener.
	 * 
	 * @param key
	 */
	private void doRead(SelectionKey key)
	{
		RpcConnection connection = (RpcConnection) key.attachment();
		int readCount = -1;
		try
		{
			readCount = connection.read();
		} catch (IOException e)
		{
		}
		if (readCount == -1)
		{
			try
			{
				connection.close();
			} catch (IOException e)
			{
				// Do nothing about this exception.
			}
			connectionManager.remove(connection);
			LOGGER.info(connection.toString() + " removed!");
		}
		
	}
	
	/**
	 * Handling write event of this listener.
	 * 
	 * @param key
	 */
	private void doWrite(SelectionKey key)
	{
		RpcConnection connection = (RpcConnection) key.attachment();
		try
		{
			connection.write();
		} catch (IOException e)
		{
			try
			{
				connection.close();
			} catch (IOException e1)
			{
				// Do nothing about this exception.
			}
			connectionManager.remove(connection);
			LOGGER.info(connection.toString() + " removed!");
		}
	}
	
	/**
	 * Initialize this listener, also start this thread.
	 * 
	 * @throws IOException
	 *             If can not open the selector.
	 */
	public void initialize() throws IOException
	{
		this.selector = Selector.open();
		this.start();
	}
	
	/**
	 * Quit this listener thread.
	 */
	void quit()
	{
		this.running = false;
	}
	
	/**
	 * Run.
	 */
	@Override
	public void run()
	{
		
		LOGGER.info(getName() + " starting...");
		SelectionKey key;
		Iterator<SelectionKey> iterator;
		
		while (running)
		{
			try
			{
				selector.select();
				lock.lock();
				try
				{
					// try to wait pending accept tasks.
					while (pendingTaskCount > 0)
					{
						nonePendingTasks.await();
					}
					for (iterator = selector.selectedKeys().iterator(); iterator.hasNext();)
					{
						
						key = iterator.next();
						iterator.remove();
						if (key.isValid())
						{
							if (key.isReadable())
								doRead(key);
							else if (key.isWritable())
								doWrite(key);
							else
							{
								LOGGER.log(Level.SEVERE,
										"Listener got un-excepted key: " + key.toString());
							}
						}
					}
				} finally
				{
					lock.unlock();
				}
			} catch (Throwable e)
			{
				// protect this thread not to quit.
				LOGGER.log(Level.SEVERE, "Server listener got error", e);
			}
			
		}
		LOGGER.info(getName() + " stopped");
		
	}
}
